package com.example.kafka.service.impl;

import java.util.Arrays;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFuture;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.NewTopic;

import com.example.kafka.service.KafkaService;



/**
 * @author jiyupeng
 */
@Service
public class KafkaServiceImpl implements KafkaService {
	
	@Autowired 
	private KafkaTemplate<Integer, String> kafkaTemplate;
	@Autowired
	private AdminClient adminClient;

	/**
	 * 异步方式发送消息
	 */
	@Override
	@Transactional
	public void sendMessageInAsyncWay(String message) {
		//这里发送之后不会等待，返回响应
		ListenableFuture<SendResult<Integer, String>> future = kafkaTemplate.send("topic1", message);
		//可以在future中回调函数中处理成功，或者失败信息
		future.addCallback(result -> {
			System.out.println("success");
			System.out.println(result);
		}, th -> {
			System.out.println("falure");
			th.printStackTrace();
		});
	}
	
	/**
	 * 同步方式发送消息
	 */
	@Override
	public void sendMessageInSyncWay(String message) {
		try {
			//发送完消息之后等待，在Future.get()方法中设置等待时间，超时则
			kafkaTemplate.send("topic1", message).get(10, TimeUnit.SECONDS);
			System.out.println("success");
		} catch (InterruptedException | ExecutionException | TimeoutException e) {
			System.out.println("failure");
			e.printStackTrace();
		}
	}
	
	/**
	 * 使用事务发送消息
	 * @param message
	 */
	@Override
	public void sendMessageInTransaction(String message) {
		kafkaTemplate.executeInTransaction((kafkaOperations) -> {
			kafkaOperations.send("topic1", message).addCallback(result -> {
				System.out.println("success");
				System.out.println(result);
			}, th -> {
				System.out.println("falure");
				th.printStackTrace();
			});
			throw new RuntimeException();
			//return null;
		});
	}
	
	/**
	 * 使用事务注解发送消息
	 * @param message
	 * @throws Exception 
	 */
	@Override
	@Transactional
	public void sendMessageInTransactionAnno(String message) throws Exception {
		ListenableFuture<SendResult<Integer, String>> future = kafkaTemplate.send("topic1", message);
		future.addCallback(result -> {
			System.out.println("success");
			System.out.println(result);
		}, th -> {
			System.out.println("failure");
			th.printStackTrace();
		});
		throw new Exception("失败");
	}

	/**
	 * 使用AdminClient来创建topic
	 */
	@Override
	public void createTopic(String topic) {
		NewTopic newTopic = new NewTopic("topic.initial", 2, (short)1);
		adminClient.createTopics(Arrays.asList(newTopic));
		System.out.println("创建成功");
	}
	
}
